import dask.dataframe as dd
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
import seaborn as sns
import argparse
import glob
import json
import os
import re
from datetime import datetime
from pyecharts.charts import Map, Pie, Page
from pyecharts.components import Table
from pyecharts import options as opts
from pyecharts.options import ComponentTitleOpts
from matplotlib.font_manager import FontProperties

# 设置全局字体
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 指定中文字体
font = FontProperties(fname=r'C:/Windows/Fonts/simhei.ttf', size=12)

# 设置全局样式
sns.set_style('whitegrid')
OUTPUT_DIR = './output'


# 用于记录上次处理的文件列表
processed_files = []

def process_parquet_files(directory):
    """
    处理指定目录下的所有parquet文件，支持增量式输入
    """
    global processed_files
    print(f"开始处理目录: {directory}")
    
    # 检查目录是否存在
    if not os.path.exists(directory):
        raise ValueError(f"目录 {directory} 不存在")
    
    # 获取所有parquet文件
    all_parquet_files = glob.glob(os.path.join(directory, "*.parquet"))
    
    # 找出未处理的文件
    new_parquet_files = [file for file in all_parquet_files if file not in processed_files]
    
    # 检查是否找到新文件
    if not new_parquet_files:
        print("没有找到新的parquet文件，无需处理。")
        return None
    
    print(f"找到 {len(new_parquet_files)} 个新的parquet文件")
    print("新文件列表：")
    for file in new_parquet_files:
        print(f"- {file}")
    
    print("开始读取新的parquet文件...")
    try:
        df = dd.read_parquet(new_parquet_files)
        print("parquet文件读取完成，开始计算...")
        # 转换为pandas DataFrame
        df = df.compute()
        print(f"新数据处理完成，总行数: {len(df)}")
        
        # 更新已处理文件列表
        processed_files.extend(new_parquet_files)
        
        return df
    except Exception as e:
        print(f"读取parquet文件时出错: {str(e)}")
        raise

def visualize_age_distribution(df, output_dir):
    """
    可视化年龄分布
    """
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    
    # 直方图
    plt.figure(figsize=(10, 6))
    sns.histplot(df['age'], bins=20, kde=True)
    plt.title('年龄分布', fontproperties=my_font)
    plt.xlabel('年龄', fontproperties=my_font)
    plt.ylabel('频数', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'age_histogram.png'))
    plt.close()

    # 箱线图
    plt.figure(figsize=(10, 6))
    sns.boxplot(x=df['age'])
    plt.title('年龄箱线图', fontproperties=my_font)
    plt.xlabel('年龄', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'age_boxplot.png'))
    plt.close()

def check_email_outliers(df):
    """
    检查email列中的离群值
    """
    # 定义email正则表达式
    email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    
    # 检查不符合email格式的值
    invalid_emails = df[~df['email'].str.match(email_pattern, na=False)]
    
    return len(invalid_emails)

def visualize_income_distribution(df, output_dir):
    """
    可视化收入分布
    """
    # 检查负值
    negative_income_count = len(df[df['income'] < 0])
    print(f"收入列中负值的数量: {negative_income_count}")
    
    # 直方图
    plt.figure(figsize=(10, 6))
    sns.histplot(df['income'], bins=20, kde=True)
    plt.title('收入分布', fontproperties=my_font)
    plt.xlabel('收入', fontproperties=my_font)
    plt.ylabel('频数', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'income_histogram.png'))
    plt.close()

def visualize_gender_distribution(df, output_dir):
    """
    可视化性别分布
    """
    gender_counts = df['gender'].value_counts()
    
    # 饼图
    plt.figure(figsize=(8, 8))
    plt.pie(gender_counts, labels=gender_counts.index, autopct='%1.1f%%', textprops={'fontproperties': my_font})
    plt.title('性别比例', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'gender.png'))
    plt.close()

    # 条形图
    plt.figure(figsize=(10, 6))
    ax = sns.barplot(x=gender_counts.index, y=gender_counts.values)
    plt.title('性别分布', fontproperties=my_font)
    plt.xlabel('性别', fontproperties=my_font)
    plt.ylabel('人数', fontproperties=my_font)
    
    # 设置x轴刻度的字体
    plt.xticks(fontproperties=my_font)
    # 设置y轴刻度的字体
    plt.yticks(fontproperties=my_font)
    
    plt.savefig(os.path.join(output_dir, 'gender_bar.png'))
    plt.close()

def extract_province(address):
    """
    从地址中提取省份
    支持的省级行政区类型：
    - 省（如：广东省）
    - 自治区（如：广西壮族自治区）
    - 直辖市（如：北京市）
    - 特别行政区（如：香港特别行政区）
    """
    # 定义省级行政区模式
    patterns = [
        r'^(.*?特别行政区)',  # 特别行政区
        r'^(.*?自治区)',      # 自治区
        r'^(.*?省)',          # 省
        r'^(.*?市)'           # 直辖市
    ]
     # 首先检查是否为非中国地址
    if "Non-Chinese Address Placeholder" in address:
        print(f"非中国地址: {address}")
        return "非中国地址"
    # 按优先级依次匹配
    for pattern in patterns:
        match = re.match(pattern, address)
        if match:
            return match.group(1)
    
    # 如果没匹配到，打印出这个地址以便调试
    print(f"未匹配的地址: {address}")
    return "未知"

def visualize_province_distribution(df, output_dir):
    """
    可视化省级行政区分布
    """
    print("开始提取省级行政区信息...")
    df['province'] = df['address'].apply(extract_province)
    
    # 统计并打印结果
    province_counts = df['province'].value_counts()
    province_ratios = province_counts / province_counts.sum()
    
    print("省级行政区统计结果：")
    for province, count in province_counts.items():
        print(f"{province}: {count}条")
    
    # ------------------ Map 地图 ------------------
    data_pair = list(zip(province_counts.index.tolist(), province_counts.tolist()))
    map_chart = (
        Map()
        .add("地址分布", data_pair, "china")
        .set_global_opts(
            title_opts=opts.TitleOpts(title="省级地址分布地图", pos_left="center"),
            visualmap_opts=opts.VisualMapOpts(max_=province_counts.max())
        )
    )

    # ------------------ 优化后的饼图 ------------------
    pie_data = sorted(
        [list(z) for z in zip(province_counts.index.tolist(), province_ratios.tolist())],
        key=lambda x: x[1],
        reverse=True
    )
    
    pie_chart = (
        Pie(init_opts=opts.InitOpts(width="1000px", height="600px"))
        .add(
            series_name="人数占比",
            data_pair=pie_data,
            radius=["30%", "70%"],
            center=["50%", "50%"],
            rosetype="radius",  # 使用南丁格尔玫瑰图
            label_opts=opts.LabelOpts(
                is_show=True,
                formatter="{b}: {d}%",
                position="outside",
                font_size=12,
                color="#333"
            ),
        )
        .set_global_opts(
            title_opts=opts.TitleOpts(
                title="各省人数占比",
                pos_left="center",
                title_textstyle_opts=opts.TextStyleOpts(font_size=16)
            ),
            legend_opts=opts.LegendOpts(
                orient="vertical",
                pos_top="15%",
                pos_left="2%",
                item_width=10,
                item_height=10,
                textstyle_opts=opts.TextStyleOpts(font_size=10)
            ),
            tooltip_opts=opts.TooltipOpts(
                trigger="item",
                formatter="{a}<br/>{b}: {c} ({d}%)"
            )
        )
        .set_series_opts(
            tooltip_opts=opts.TooltipOpts(
                formatter="{b}: {c} ({d}%)"
            )
        )
    )

    # ------------------ 表格 Table ------------------
    table = Table()
    headers = ["省份", "人数", "占比"]
    rows = [[prov, str(province_counts[prov]), f"{province_ratios[prov]:.2%}"]
            for prov in province_counts.index]
    table.add(headers, rows).set_global_opts(
        title_opts=ComponentTitleOpts(title="各省人数详情表格")
    )

    # ------------------ 页面组合 ------------------
    page = Page(layout=Page.DraggablePageLayout)  # 使用可拖拽布局
    page.add(
        map_chart,
        pie_chart,
        table
    )
    page.render(os.path.join(output_dir, 'province_distribution.txt'))
    print("省级行政区分布图表已生成")

def visualize_active_status(df, output_dir):
    """
    可视化活跃状态分布
    """
    active_counts = df['is_active'].value_counts()
    
    plt.figure(figsize=(8, 8))
    plt.pie(active_counts, labels=active_counts.index, autopct='%1.1f%%', textprops={'fontproperties': my_font})
    plt.title('用户活跃状态比例', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'active_status.png'))
    plt.close()

def visualize_registration_trend(df, output_dir):
    """
    可视化注册时间趋势
    """
    # 转换日期格式
    df['registration_date'] = pd.to_datetime(df['registration_date'])
    
    # 年度注册趋势
    df['registration_year'] = df['registration_date'].dt.year
    yearly_registration = df['registration_year'].value_counts().sort_index()
    
    plt.figure(figsize=(12, 6))
    sns.lineplot(x=yearly_registration.index, y=yearly_registration.values)
    plt.title('年度注册趋势', fontproperties=my_font)
    plt.xlabel('年份', fontproperties=my_font)
    plt.ylabel('注册人数', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'yearly_registration_trend.png'))
    plt.close()
    
    # 日度注册趋势
    daily_registration = df['registration_date'].value_counts().sort_index()
    
    plt.figure(figsize=(12, 6))
    sns.lineplot(x=daily_registration.index, y=daily_registration.values)
    plt.title('日度注册趋势', fontproperties=my_font)
    plt.xlabel('日期', fontproperties=my_font)
    plt.ylabel('注册人数', fontproperties=my_font)
    plt.xticks(rotation=45)
    plt.savefig(os.path.join(output_dir, 'daily_registration_trend.png'))
    plt.close()

def visualize_country_distribution(df, output_dir):
    """
    可视化国家分布
    """
    country_counts = df['country'].value_counts()
    
    # 饼图
    plt.figure(figsize=(12, 8))
    plt.pie(country_counts, labels=country_counts.index, autopct='%1.1f%%', 
            textprops={'fontproperties': my_font})
    plt.title('国家分布比例', fontproperties=my_font)
    plt.savefig(os.path.join(output_dir, 'country_distribution.png'))
    plt.close()

    # 条形图
    plt.figure(figsize=(15, 8))
    ax = sns.barplot(x=country_counts.index, y=country_counts.values)
    plt.title('国家分布', fontproperties=my_font)
    plt.xlabel('国家', fontproperties=my_font)
    plt.ylabel('人数', fontproperties=my_font)
    
    # 设置x轴刻度的字体和旋转角度
    plt.xticks(rotation=45, fontproperties=my_font)
    # 设置y轴刻度的字体
    plt.yticks(fontproperties=my_font)
    
    # 调整布局以防止标签被截断
    plt.tight_layout()
    
    plt.savefig(os.path.join(output_dir, 'country_bar.png'))
    plt.close()


def check_timestamp_anomalies(df):
    """
    检查并修正时间戳相关的异常值：
    - 统一时间格式和时区
    - 检查last_login是否早于registration_date并修正
    - 检查时间戳是否在未来并修正
    """
    print("\n=== 时间戳异常值检查和修正 ===")
    date_columns = ['last_login', 'registration_date']
    current_time = pd.Timestamp.now(tz='UTC')  # 使用UTC时区
    
    # 确保时间列已转换为datetime格式并统一时区
    for col in date_columns:
        if col in df.columns:
            if df[col].dtype != 'datetime64[ns, UTC]':
                print(f"将{col}转换为datetime格式并统一时区...")
                # 先转换为datetime
                df[col] = df[col].map_partitions(
                    pd.to_datetime,
                    meta=pd.Series([], dtype='datetime64[ns]')
                )
                # 然后统一时区
                df[col] = df[col].map_partitions(
                    lambda x: x.dt.tz_localize('UTC') if x.dt.tz is None else x.dt.tz_convert('UTC'),
                    meta=pd.Series([], dtype='datetime64[ns, UTC]')
                )
    
    # 1. 检查并修正last_login早于registration_date的情况
    if 'last_login' in df.columns and 'registration_date' in df.columns:
        # 使用Dask的延迟计算
        anomaly_mask = df['last_login'] < df['registration_date']
        anomaly_count = anomaly_mask.sum().compute()
        print(f"last_login早于registration_date的行数: {anomaly_count}")
        
        if anomaly_count > 0:
            print("示例数据:")
            # 使用Dask的head()方法获取示例数据
            sample_data = df[anomaly_mask].head()
            # 确保sample_data是Dask DataFrame
            if isinstance(sample_data, dd.DataFrame):
                sample_data = sample_data[['last_login', 'registration_date']].compute()
            print(sample_data)
            print("\n修正异常值: 将last_login设置为registration_date")
            # 修正异常值
            df = df.map_partitions(
                lambda x: x.assign(
                    last_login=lambda df: df.apply(
                        lambda row: row['registration_date'] if row['last_login'] < row['registration_date'] else row['last_login'],
                        axis=1
                    )
                ),
                meta=df._meta
            )
    
    
    
    return df

def analyze_data_quality(df):
    """
    分析数据质量，包括缺失值、异常值等
    """
    print("\n=== 数据质量分析 ===")
    
    # 1. 缺失值分析
    missing_stats = df.isnull().sum().compute()
    missing_percentage = (missing_stats / len(df)) * 100
    missing_info = pd.DataFrame({
        '缺失值数量': missing_stats,
        '缺失值比例(%)': missing_percentage
    })
    print("\n缺失值统计:")
    print(missing_info)
    
    # 2. 数据类型分析
    print("\n数据类型分析:")
    print(df.dtypes)
    
    # 3. 时间戳异常值检查和修正
    df = check_timestamp_anomalies(df)
    
   
    
    return df, missing_info



def calculate_rfm_metrics(pdf, NOW):
    """
    计算 RFM 指标
    """
    pdf['Recency'] = (NOW - pdf['last_login']).dt.days.astype('int32')
    
    def calculate_purchase_frequency(history):
        try:
            if isinstance(history, str):
                purchase_data = json.loads(history)
                return len(purchase_data.get('items', []))
            return 0
        except:
            return 0
    
    pdf['Frequency'] = pdf['purchase_history'].apply(calculate_purchase_frequency).astype('int32')
    
    def calculate_value(row):
        try:
            value = row['income'] * 0.7
            age_factor = 1.2 if 25 <= row['age'] <= 45 else 0.8
            return value * age_factor
        except:
            return row['income']
    
    pdf['Monetary'] = pdf.apply(calculate_value, axis=1).astype('float32')
    return pdf

def score_rfm(rfm_df):
    """
    对 RFM 指标进行评分
    """
    rfm_df['R_Score'] = pd.qcut(rfm_df['Recency'].rank(method='first'), 5, labels=range(5, 0, -1)).astype('int8')
    rfm_df['F_Score'] = pd.qcut(rfm_df['Frequency'].rank(method='first'), 5, labels=range(1, 6)).astype('int8')
    rfm_df['M_Score'] = pd.qcut(rfm_df['Monetary'].rank(method='first'), 5, labels=range(1, 6)).astype('int8')
    return rfm_df

def analyze_rfm(df, output_dir):
    """
    进行RFM高价值用户分析，并生成可视化报告
    """
    print("\n=== 开始RFM高价值用户分析 ===")
    print("计算RFM指标...")
    
    pdf = df[['id', 'user_name', 'last_login', 'purchase_history', 'income', 'age']].compute()
    NOW = pd.Timestamp.now(tz='UTC')
    
    pdf = calculate_rfm_metrics(pdf, NOW)
    rfm_df = pdf[['Recency', 'Frequency', 'Monetary']].copy()
    rfm_df = score_rfm(rfm_df)
    
    # 后续可添加生成可视化报告的子函数调用
    
    return rfm_df

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input_dir", type=str, default="./data/10G_data")
    parser.add_argument("--output_dir", type=str, default="./output")
    args = parser.parse_args()
    
    print("程序开始运行...")
    print(f"输入目录: {args.input_dir}")
    print(f"输出目录: {args.output_dir}")
    
    start_time = datetime.now()
    print(f"开始时间: {start_time}")
    
    try:
        # 处理数据
        print("开始处理数据...")
        df = process_parquet_files(args.input_dir)
        
        # 创建输出目录
        if not os.path.exists(args.output_dir):
            os.makedirs(args.output_dir)
            print(f"创建输出目录: {args.output_dir}")
        
        # 执行各项分析
        print("开始生成年龄分布图表...")
        visualize_age_distribution(df, args.output_dir)
        
        print("开始检查邮箱格式...")
        invalid_email_count = check_email_outliers(df)
        print(f"无效邮箱数量: {invalid_email_count}")
        
        print("开始生成收入分布图表...")
        visualize_income_distribution(df, args.output_dir)
        
        print("开始生成性别分布图表...")
        visualize_gender_distribution(df, args.output_dir)
        
        print("开始生成国家分布图表...")
        visualize_country_distribution(df, args.output_dir)
        
        print("开始生成省级行政区分布图表...")
        visualize_province_distribution(df, args.output_dir)
        
        print("开始生成活跃状态图表...")
        visualize_active_status(df, args.output_dir)
        
        print("开始生成注册时间趋势图表...")
        visualize_registration_trend(df, args.output_dir)

                # 数据质量分析和时间戳修正
        df, missing_info = analyze_data_quality(df)
        
        # 如果需要，进行RFM分析
        if args.analyze_rfm:
            rfm_df = analyze_rfm(df, args.output_dir)
        
        # 保存数据质量报告到控制台
        print("\n=== 数据质量报告 ===")
        print("缺失值统计:")
        print(missing_info)
        print("\n数据类型分析:")
        print(df.dtypes)
        
        end_time = datetime.now()
        print(f"结束时间: {end_time}")
        print(f"总耗时: {end_time - start_time}")
        
        # 保存时间信息到控制台
        print(f"开始时间: {start_time}")
        print(f"结束时间: {end_time}")
        print(f"总耗时: {end_time - start_time}")
            
        print("所有分析完成！")
        
    except Exception as e:
        print(f"程序运行出错: {str(e)}")
        raise

if __name__ == "__main__":
    main()